package cn.qdhlm.frontendprocessor.core.callback;


import cn.qdhlm.frontendprocessor.common.helper.MqttHelper;
import cn.qdhlm.frontendprocessor.common.helper.RocketMqProducer;
import cn.qdhlm.frontendprocessor.common.util.CodeUtil;
import cn.qdhlm.frontendprocessor.common.util.ComUtil;
import cn.qdhlm.frontendprocessor.common.util.JsonUtil;
import cn.qdhlm.frontendprocessor.model.dto.CommandParamDTO;
import cn.qdhlm.frontendprocessor.model.mongo.Command;
import cn.qdhlm.frontendprocessor.model.rocketmq.OnlineRecord;
import com.alibaba.fastjson.JSONObject;
import io.netty.buffer.ByteBufUtil;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

/**
 * @Name:RocketMQMessageCallback
 * @description:
 * @author: xinY
 **/
@Log4j2
@Component
public class RocketMQMessageCallback implements MessageListenerConcurrently {
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private RocketMqProducer producer;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt msg : list) {
            log.info("topic:"
                    + msg.getTopic() + ", tags:"
                    + msg.getTags() + ", body:"
                    + ByteBufUtil.hexDump(msg.getBody())
                    + ", from broker:" + msg.getBrokerName()
                    + ", properties:" + msg.getProperty("KEYS")
                    + ", from queue:" + msg.getQueueId()
                    + ", message ID:" + msg.getMsgId()

            );
            // mqtt的命令
            if ("mqtt".equals(msg.getTags())) {
                sendCmd4Mqtt(msg);
            }
            // mqtt-nb的命令
            else if ("nb-mqtt".equals(msg.getTags())) {
                dealCmd4NbMqtt(msg);
            } else if ("nb-is-on".equals(msg.getTags())) {
                sendCmd4NbMqtt(msg);
            } else if ("sleep".equals(msg.getTags())) {
                sendSleep(msg);
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    /**
     * 休眠包来了
     *
     * @param msg
     */
    private void sendSleep(MessageExt msg) {
        Command command = JSONObject.parseObject(msg.getBody(), Command.class);
        System.out.println(command);
        // 发布
        String topic = "Client/" + command.getCollectorNo();
        MqttHelper.publish(topic, CodeUtil.base64StringToByteArray(command.getData()));
        // 清空这个terminalNo的所有命令
        String terminalNo = command.getTerminalNo();
        Query query = new Query();
        query.addCriteria(Criteria.where("terminalNo").is(terminalNo));
        List<Command> list = mongoTemplate.find(query, Command.class);
        for (Command thisCommand : list) {
            mongoTemplate.remove(thisCommand);
        }
    }

    /**
     * 处理  当nb来消息了
     *
     * @param msg
     */
    private void sendCmd4NbMqtt(MessageExt msg) {
        // 拿到在线记录
        OnlineRecord onlineRecord = JSONObject.parseObject(msg.getBody(), OnlineRecord.class);
        System.out.println(onlineRecord);
        // 拿到记录的terminalNo
        String terminalNo = onlineRecord.getTerminalNo();

        // 通过 terminalNo查询 未发出的命令
        Query query = new Query();
        query.addCriteria(Criteria.where("terminalNo").is(terminalNo));
        query.addCriteria(Criteria.where("status").is("READY"));
        List<Command> list = null;
        try {
            list = mongoTemplate.find(query, Command.class);
        } catch (Exception e) {
            e.printStackTrace();
        }

        String topic;
        // 如果没有命令
        if (ComUtil.isEmpty(list) || 0 >= list.size()) {
            // 休眠
            System.out.println("没有命令  休眠");
            CommandParamDTO commandParamDTO = new CommandParamDTO();
            commandParamDTO.setAfn("0c");
            commandParamDTO.setFn("2");
            commandParamDTO.setMacAddr(onlineRecord.getMacAddr());
            commandParamDTO.setCodeCustomNb(onlineRecord.getCodeCustomNb());
            // 要去组休眠包
            producer.send("DeviceCommand", "nb-mqtt", JsonUtil.safeJsonBytes(commandParamDTO));

        }
        // 如果有命令
        else {
            for (Command command : list) {
                System.out.println("有命令:" + command);

                topic = "NB_Client/" + command.getCollectorNo() + "/" + command.getMacAddr();
                // 发布
                boolean publishFlag = MqttHelper.publish(topic, CodeUtil.base64StringToByteArray(command.getData()));
                // 更新状态
                command.setUpdateTime(new Date());
                command.setSendTime(new Date());
                command.setStatus(publishFlag ? "SUCCESS" : "FAIL");
                // 保存mongo
                mongoTemplate.save(command);
            }
        }
    }

    /**
     * 来mqtt命令了
     *
     * @param msg
     */
    private void sendCmd4Mqtt(MessageExt msg) {
        // 拿到command
        Command command = JSONObject.parseObject(msg.getBody(), Command.class);
        System.out.println("command+++++>>>" + JsonUtil.safeJsonStr(command));

        String topic = "Client/" + command.getCollectorNo();
        // 发布
        boolean publishFlag = MqttHelper.publish(topic, CodeUtil.base64StringToByteArray(command.getData()));
        // 更新状态成功
        Date date = new Date();
        command.setSendTime(date);
        command.setStatus(publishFlag ? "SUCCESS" : "FAIL");
        // 保存mongo
        mongoTemplate.save(command);
    }

    /**
     * 处理nb命令包
     *
     * @param msg
     */
    private void dealCmd4NbMqtt(MessageExt msg) {
        // 拿到command
        Command command = JSONObject.parseObject(msg.getBody(), Command.class);
        // 更新状态准备
        command.setStatus("READY");
        command.setUpdateTime(new Date());
        // 保存mongo
        mongoTemplate.save(command);
    }
}
